// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

/// example:
/// bazel build //cc/appcoin2/support/feed:interval
/// ./bazel-bin/cc/appcoin2/support/feed/interval --driver ./data/coin2/feed/examples/...json

#include <boost/algorithm/string.hpp>
#include <cxxopts.hpp>

#include "coin2/app/driver.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/config/args.h"
#include "coin2/exchange/base/order/order.h"
#include "coin2/exchange/feed.pb.h"
#include "coin2/strategy/strategy.h"
#include "presto/quant/math/moving_average.h"

#include <highfive/H5File.hpp>
#include <highfive/H5DataSet.hpp>
#include <highfive/H5Group.hpp>

using coin2::exchange::base::feed::FeedUpdate;
using namespace coin2::app;
using namespace coin2::base::config;
using presto::math::TimeWindowMovingAverage;

struct ProductData {
  std::vector<int64_t> book_timestamp;
  std::vector<int64_t> trade_timestamp;

  std::vector<double> midp;
  // std::vector<double> spread;
  std::vector<double> spread_5s;
  std::vector<double> trade_qty;
  std::vector<double> trade_price;
  std::vector<double> ask_1bps_qty;
  std::vector<double> bid_1bps_qty;

  TimeWindowMovingAverage<double> spread_5s_window;

  // latency
  std::vector<double> trade_intervals_ms;
  std::vector<double> book_intervals_ms;
  int64_t last_trade_ts;
  int64_t last_book_ts;
};

struct IntervalProductData {
  int64_t start_of_day;
  std::vector<int64_t> interval_timestamp;
  std::vector<double> open_price;
  std::vector<double> high_price;
  std::vector<double> low_price;
  std::vector<double> close_price;
  std::vector<double> trade_qty;
  std::vector<int32_t> book_count;

  void Reserve(int64_t ts) {
    start_of_day = ts / (86'400'000'000'000LL) * (86'400'000'000'000LL);
    interval_timestamp.resize(86400);
    open_price.resize(86400, std::numeric_limits<double>::quiet_NaN());
    high_price.resize(86400, std::numeric_limits<double>::quiet_NaN());
    low_price.resize(86400, std::numeric_limits<double>::quiet_NaN());
    close_price.resize(86400, std::numeric_limits<double>::quiet_NaN());
    trade_qty.resize(86400);
    book_count.resize(86400);
  }
};

std::map<std::string, ProductData> data;
std::map<std::string, IntervalProductData> interval_data;

struct Strategy : public ::coin2::strategy::IStrategy {
  void onTradeFeed(const FeedUpdate& upd) override {
    auto ts = upd.timestamp();
    auto id = upd.system_name() + ":" + upd.product().absolute_norm();
    auto& trade = upd.trade();

    if (interval_data.find(id) == interval_data.end()) {
      interval_data[id].Reserve(ts);
    }

    // event_data
    {
      auto& item = data[id];

      item.trade_timestamp.push_back(ts);
      item.trade_price.push_back(trade.price);
      item.trade_qty.push_back(trade.fill_qty);
      if (item.last_trade_ts > 0) {
        auto interval = static_cast<double>(ts - item.last_trade_ts) * 1e-6;
        if (interval > 0) {
          item.trade_intervals_ms.push_back(interval);
        }
      }
      item.last_trade_ts = ts;
    }

    // interval_data
    {
      auto& item = interval_data[id];
      int64_t interval_id = (ts - item.start_of_day) / 1e9;
      item.trade_qty[interval_id] += trade.fill_qty;
    }
  }
  void onBookFeed(const FeedUpdate& upd) override {
    auto ts = upd.timestamp();
    auto id = upd.system_name() + ":" + upd.product().absolute_norm();
    auto& book = upd.book();

    if (interval_data.find(id) == interval_data.end()) {
      interval_data[id].Reserve(ts);
    }

    // event_data
    if (book.Ask0() && book.Bid0()) {
      auto& item = data[id];

      item.book_timestamp.push_back(ts);

      auto ask = book.Ask0()->price;
      auto bid = book.Bid0()->price;
      auto mid = (ask + bid) * 0.5;
      item.midp.push_back(mid);

      {
        double spread = (ask - bid) / mid * 10000.0;
        // item.spread.push_back(spread);

        if (item.spread_5s.empty()) {
          item.spread_5s_window.ResizeWindow(5'000'000'000);
        }
        item.spread_5s_window.Update(ts, spread);
        item.spread_5s.push_back(item.spread_5s_window.average());
      }

      // midp +- 1bps
      {
        double bound = book.Ask0()->price * 1.0001;
        double sum = 0;
        for (auto& [p, q] : book.Asks()) {
          if (p > bound) {
            break;
          }
          sum += q;
        }
        item.ask_1bps_qty.push_back(sum);
      }
      {
        double bound = book.Bid0()->price * 0.9999;
        double sum = 0;
        for (auto& [p, q] : book.Bids()) {
          if (p < bound) {
            break;
          }
          sum += q;
        }
        item.bid_1bps_qty.push_back(-sum);
      }

      if (item.last_book_ts > 0) {
        auto interval = static_cast<double>(ts - item.last_book_ts) * 1e-6;
        if (interval > 0) {
          item.book_intervals_ms.push_back(interval);
        }
      }
      item.last_book_ts = ts;
    }

    // interval_data
    if (book.Ask0() && book.Bid0()) {
      auto& item = interval_data[id];
      int64_t interval_id = (ts - item.start_of_day) / 1e9;
      double price = (book.Ask0()->price + book.Bid0()->price) / 2.0;
      if (std::isnan(item.open_price[interval_id]))
        item.open_price[interval_id] = price;
      if (std::isnan(item.high_price[interval_id]))
        item.high_price[interval_id] = price;
      else
        item.high_price[interval_id] = std::max(item.high_price[interval_id], price);
      if (std::isnan(item.low_price[interval_id]))
        item.low_price[interval_id] = price;
      else
        item.low_price[interval_id] = std::min(item.low_price[interval_id], price);
      item.close_price[interval_id] = price;
      item.book_count[interval_id]++;
    }
  }
};

template <typename T>
void write_dataset(
    HighFive::Group* group, const std::string& name, const std::vector<T>& data) {
  HighFive::DataSetCreateProps props;
  props.add(HighFive::Chunking(std::vector<hsize_t>{data.size()}));
  props.add(HighFive::Deflate(7));
  HighFive::DataSet dataset = group->createDataSet<T>(
      name,
      HighFive::DataSpace::From(data),
      props,
      HighFive::DataSetAccessProps());
  dataset.write(data);
}


using coin2::app::DriverConfig;

void parse_cli(int argc, char* argv[], DriverConfig* app, std::string* output) {
  // you have 2 options to provide flags.
  // 1. --driver
  // 2. --machine --fastfeed --...
  // mixing these 2 is discouraged.

  cxxopts::Options opt("Feed Plot", "Plot feed");
  AddDriverOptions(&opt);                                         // --driver
  AddIntervalOptions(&opt);                                       // --start --end --duration
  opt.add_options()                                               // add options
      ("fastfeed", "use fastfeed")                                //
      ("raw", "use raw")                                          //
      ("machine", "feed machine", cxxopts::value<std::string>())  //
      ("mea", "M.E.A", cxxopts::value<std::string>())             //
      ("recipe", "recipe", cxxopts::value<std::string>())         //
      ("product", "a norm symbol", cxxopts::value<std::vector<std::string>>())
      ("output", "output filename", cxxopts::value<std::string>()->default_value("output.h5"));

  auto res = opt.parse(argc, argv);

  if (!ParseDriverOptions(res, app)) {
    auto& exchanges = *(app->mutable_feed()->mutable_exchanges());
    auto& exchange = exchanges[res["mea"].as<std::string>()];

    exchange.set_mea(res["mea"].as<std::string>());
    exchange.set_recipe(res["recipe"].as<std::string>());
    exchange.set_type(coin2::exchange::feed::FeedSystemType::FEED_SYSTEM_TYPE_ARCHIVE);
    for (const auto& norm : res["product"].as<std::vector<std::string>>()) {
      exchange.mutable_products()->add_norms(norm);
    }

    if (!(res["fastfeed"].as<bool>() ^ res["raw"].as<bool>())) {
      THROW() << "You should provide only one of --fastfeed or --raw";
    }
    bool use_fastfeed = res["fastfeed"].as<bool>();
    exchange.mutable_archive()->set_archive_type(
        use_fastfeed ? coin2::exchange::feed::ArchiveConfig::FASTFEED
                     : coin2::exchange::feed::ArchiveConfig::RAW);
    exchange.mutable_archive()->set_machine(res["machine"].as<std::string>());
  }
  ParseIntervalOptions(
      res,
      app->mutable_feed()->mutable_common()->mutable_archive()->mutable_interval());

  *output = res["output"].as<std::string>();
}

int main(int argc, char* argv[]) {
  /* ----------------------------------- opt ---------------------------------- */
  coin2::app::DriverConfig app;
  std::string output;  // extract this variable later.
  parse_cli(argc, argv, &app, &output);

  /* ----------------------------------- run ---------------------------------- */
  printf("reading ..\n");

  Strategy strategy;
  ::coin2::app::Driver driver(app, &strategy);
  driver.Run();

  /* ---------------------------------- dump ---------------------------------- */
  printf("Dumping %lu products to %s ..", data.size(), output.c_str());

  HighFive::File file(
      output,
      HighFive::File::ReadWrite | HighFive::File::Create | HighFive::File::Truncate);

  /* ---------------------------- dump ----------- */
  {
    for (auto& [key, item] : data) {
      HighFive::Group group = file.createGroup(key + ":event");
      (void)group;
      (void)item;
      write_dataset<int64_t>(&group, "book_timestamp", item.book_timestamp);
      write_dataset<double>(&group, "midp", item.midp);
      write_dataset<int64_t>(&group, "trade_timestamp", item.trade_timestamp);
      write_dataset<double>(&group, "tradep", item.trade_price);
      write_dataset<double>(&group, "tradeq", item.trade_qty);
    }
  }

  {
    for (auto& [key, item] : interval_data) {
      HighFive::Group group = file.createGroup(key + ":interval");
      (void)group;
      write_dataset<int64_t>(&group, "interval_timestamp", item.interval_timestamp);
      write_dataset<double>(&group, "open_price", item.open_price);
      write_dataset<double>(&group, "high_price", item.high_price);
      write_dataset<double>(&group, "low_price", item.low_price);
      write_dataset<double>(&group, "close_price", item.close_price);
      write_dataset<double>(&group, "trade_qty", item.trade_qty);
      write_dataset<int32_t>(&group, "book_count", item.book_count);
    }
  }

  return 0;
}
